package cn.doitedu;

import com.clearspring.analytics.stream.cardinality.HyperLogLog;
import org.apache.hadoop.hive.ql.exec.UDAF;
import org.apache.hadoop.hive.ql.exec.UDAFEvaluator;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.io.IOException;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/4/24
 * @Desc: 学大数据，上多易教育
 *   把一组整数聚合成一个bitmap
 **/
public class AggStringToHLLUDAF extends UDAF {


    public static class AggBigIntToBitmapEvaluator implements UDAFEvaluator {

        HLLAcc acc;

        public AggBigIntToBitmapEvaluator() {
            init();
        }


        // 初始化中间累加器
        @Override
        public void init() {
            try {
                acc = new HLLAcc();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }



        // 逐条聚合
        public boolean iterate(String element) throws IOException, ClassNotFoundException {

            // 取出累加器中的hll
            HyperLogLog hyperLogLog = this.acc.getHyperLogLog();


            // 向hll中添加元素
            hyperLogLog.offer(element);


            // 把添加元素后的hll重新设置到累加器中
            this.acc.setHyperLogLog(hyperLogLog);

            return true;
        }


        // map端输出的局部结果
        public HLLAcc terminatePartial() {
            return this.acc;
        }


        // reduce端合并map端输出的acc
        public boolean merge(HLLAcc otherAcc) throws Exception {

            // 取出自己的累加器中的 hll
            HyperLogLog thisHll = this.acc.getHyperLogLog();

            // 取出传入的另一个累加器中的 hll
            HyperLogLog thatHll = otherAcc.getHyperLogLog();


            // 合并两个hll到thisBitmap
            thisHll.addAll(thatHll);


            // 把合并后的hll重新设置到累加器中
            this.acc.setHyperLogLog(thisHll);

            return true;
        }

        // 输出最终结果
        public byte[] terminate() {
            return this.acc.getHllBytes();
        }
    }

}
